package vertxRestZeebeKafka;

import io.vertx.core.Promise;
import io.vertx.core.json.JsonObject;
import io.vertx.kafka.client.producer.KafkaProducer;
import io.vertx.kafka.client.producer.KafkaProducerRecord;
import io.zeebe.client.ZeebeClient;
import io.zeebe.client.api.response.ActivatedJob;
import io.zeebe.client.api.worker.JobClient;
import io.zeebe.client.api.worker.JobHandler;

public class PublishJobWorkerHandler implements JobHandler {
    ZeebeClient zeebeClient;
    KafkaProducer<Long, JsonObject> producer;

    public PublishJobWorkerHandler(ZeebeClient client, KafkaProducer<Long, JsonObject> producer) {
        this.zeebeClient = client;
        this.producer = producer;
    }

    @Override
    public void handle(JobClient client, ActivatedJob job) throws Exception {
        System.out.println("开始执行publishJobWorker");

        JsonObject data = new JsonObject(job.getVariables());
        System.out.println("publishJobWorker执行完成。output=" + data);


        long key = job.getWorkflowInstanceKey();
        Promise<JsonObject> promise = Promise.promise();
        producer.write(KafkaProducerRecord.create("zeebeOutput", key, data), ar -> {
            if (ar.succeeded()) {
                client.newCompleteCommand(job.getKey())
                        .send()
                        .join();
                System.out.println("工作流实例key=" + key + ",kafka发送output的topic的消息完成");

                promise.complete();
            } else {
                client.newFailCommand(job.getKey());
                System.out.println("工作流实例key=" + key + ",kafka发送output的topic的消息失败");

                promise.fail(ar.cause());
            }
        });
    }
}
